#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
// 引入muduo网络库的TCP连接类，用于管理与服务器的TCP连接
#include "muduo/net/TcpConnection.h"
// 引入muduo的Protobuf编解码器和消息分发器，用于序列化/反序列化Protobuf消息
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
// 引入自定义的日志工具（DLOG宏）、UUID生成工具（UUIDHelper）
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
// 引入自定义的Protobuf消息定义，包含消息队列的核心数据结构
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
// 引入消费者类，用于处理消息订阅和回调逻辑
#include "mq_consumer.hpp"
// 引入标准库头文件：输入输出、线程同步、哈希表
#include <iostream>
#include <mutex>
#include <condition_variable>
#include <unordered_map>

namespace zdfmq
{
    // 定义Protobuf消息的智能指针类型别名，统一管理消息对象生命周期
    typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
    // 定义Protobuf编解码器的智能指针类型，避免原始指针的内存管理问题
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    // 定义特定响应消息的智能指针，简化代码中指针的声明
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;

    /**
     * @brief 信道类（Channel）是消息队列客户端与服务器交互的核心逻辑单元
     * 设计思路：基于AMQP协议的信道模型，一个TCP连接上可创建多个Channel，实现逻辑隔离
     * 核心功能：封装交换机/队列的声明、绑定、消息发布/消费等协议操作，通过请求-响应模式与服务器通信
     */
    class Channel
    {
    public:
        // 定义Channel的智能指针类型，所有外部访问均通过智能指针，确保内存安全
        using ptr = std::shared_ptr<Channel>;

        /**
         * @brief 构造函数：初始化信道ID、关联TCP连接和编解码器
         * @param conn 已建立的TCP连接指针（由客户端连接管理器传入）
         * @param codec Protobuf编解码器指针（全局唯一，由客户端初始化时创建）
         * 细节：信道ID（_cid）通过UUIDHelper生成，确保全局唯一，用于服务器区分不同信道
         */
        Channel(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec) : _cid(UUIDHelper::uuid()), _conn(conn), _codec(codec) {}

        /**
         * @brief 析构函数：自动取消当前信道的消费者订阅，避免资源泄露
         * 注意：析构时会调用basicCancel()，确保与服务器的订阅状态同步
         */
        ~Channel() { basicCancel(); }

        /**
         * @brief 获取当前信道的唯一标识
         * @return 字符串形式的信道ID（_cid）
         * 用途：在多信道场景下，用于标识当前操作所属的信道
         */
        std::string cid() { return _cid; }

        /**
         * @brief 打开信道：向服务器发送信道打开请求，完成握手流程
         * 协议流程：
         * 1. 生成唯一请求ID（rid），用于匹配请求和响应
         * 2. 构造openChannelRequest消息，设置rid和cid
         * 3. 通过编解码器发送消息到服务器
         * 4. 阻塞等待服务器响应，根据响应结果返回成功/失败
         * @return 服务器返回成功（ok()为true）则返回true，否则返回false
         * 注意：此方法会阻塞当前线程，直到收到响应或连接中断
         */
        bool openChannel()
        {
            // 生成请求ID：采用UUIDv4格式，确保分布式环境下的唯一性
            std::string rid = UUIDHelper::uuid();
            openChannelRequest req;
            req.set_rid(rid);  // 设置请求ID，用于后续响应匹配
            req.set_cid(_cid); // 绑定当前信道ID
            // 通过编解码器发送消息：内部会调用TCP连接的send方法，线程安全
            _codec->send(_conn, req);
            // 等待响应：内部通过条件变量等待，直到收到对应rid的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok(); // 返回服务器响应的状态
        }

        /**
         * @brief 关闭信道：向服务器发送关闭请求，释放相关资源
         * 协议流程：
         * 1. 生成rid，构造closeChannelRequest消息
         * 2. 发送消息后阻塞等待响应（不关心响应结果，确保服务器已处理）
         * 副作用：关闭后当前信道的所有操作（如发布消息、消费）将失效
         */
        void closeChannel()
        {
            std::string rid = UUIDHelper::uuid();
            closeChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn, req);
            waitResponse(rid); // 等待服务器确认关闭，避免提前释放资源
            return;
        }

        /**
         * @brief 声明交换机：在服务器端创建或验证交换机配置
         * 协议背景：交换机是消息路由的核心组件，必须先声明才能使用
         * @param name 交换机名称（唯一标识）
         * @param type 交换机类型（如DIRECT、TOPIC、FANOUT，定义在mq_proto.proto中）
         * @param durable 是否持久化（服务器重启后不丢失）
         * @param auto_delete 当最后一个绑定解除时是否自动删除
         * @param args 附加参数（如备用交换机、消息TTL等扩展属性）
         * @return 服务器确认声明成功则返回true
         * 注意：
         * - 若交换机已存在，服务器会验证参数是否匹配，不匹配则返回失败
         * - args参数通过swap转移所有权，避免不必要的拷贝（性能优化）
         */
        bool declareExchange(
            const std::string &name,
            ExchangeType type,
            bool durable,
            bool auto_delete,
            google::protobuf::Map<std::string, std::string> &args)
        {
            std::string rid = UUIDHelper::uuid();
            declareExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            // 将外部传入的args转移到请求消息中（通过mutable_args()获取内部map的引用）
            req.mutable_args()->swap(args);
            _codec->send(_conn, req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        /**
         * @brief 删除交换机：删除服务器上指定名称的交换机
         * 注意：
         * - 若交换机不存在，服务器可能返回失败，但此方法不处理（由上层决定是否重试）
         * - 删除前需确保交换机没有绑定关系，否则可能删除失败
         */
        void deleteExchange(const std::string &name)
        {
            std::string rid = UUIDHelper::uuid();
            deleteExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            _codec->send(_conn, req);
            waitResponse(rid); // 等待删除完成，确保状态同步
            return;
        }

        /**
         * @brief 声明队列：在服务器端创建或验证队列配置
         * 队列属性说明：
         * - durable：持久化（服务器重启后队列元数据不丢失，但消息是否持久化由发布时决定）
         * - exclusive：排他性（仅当前信道可见，信道关闭后自动删除）
         * - auto_delete：当最后一个消费者取消订阅后自动删除
         * @param qname 队列名称（为空时服务器会生成随机名称）
         * @param qdurable 是否持久化
         * @param qexclusive 是否排他性
         * @param qauto_delete 是否自动删除
         * @param qargs 附加参数（如队列长度限制、消息过期时间等）
         * @return 声明成功返回true
         */
        bool declareQueue(
            const std::string &qname,
            bool qdurable,
            bool qexclusive,
            bool qauto_delete,
            google::protobuf::Map<std::string, std::string> &qargs)
        {
            std::string rid = UUIDHelper::uuid();
            declareQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_durable(qdurable);
            req.set_auto_delete(qauto_delete);
            req.set_exclusive(qexclusive);
            req.mutable_args()->swap(qargs); // 转移参数所有权，减少拷贝
            _codec->send(_conn, req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        /**
         * @brief 删除队列：删除服务器上指定名称的队列
         * 注意：
         * - 队列必须为空（无消息）才能删除，否则服务器返回失败
         * - 若队列不存在，此操作可能无效果（取决于服务器实现）
         */
        void deleteQueue(const std::string &qname)
        {
            std::string rid = UUIDHelper::uuid();
            deleteQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }

        /**
         * @brief 绑定队列到交换机：建立队列与交换机的关联关系，指定路由键
         * 路由逻辑：消息发布到交换机后，根据路由键和交换机类型路由到绑定的队列
         * @param ename 交换机名称（必须已声明）
         * @param qname 队列名称（必须已声明）
         * @param key 路由键（根据交换机类型有不同含义，如DIRECT类型精确匹配）
         * @return 绑定成功返回true
         */
        bool queueBind(
            const std::string &ename,
            const std::string &qname,
            const std::string &key)
        {
            std::string rid = UUIDHelper::uuid();
            queueBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(key); // 路由键，用于消息路由
            _codec->send(_conn, req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        /**
         * @brief 解除队列与交换机的绑定：删除指定队列与交换机的所有绑定关系
         * 注意：此方法会解除该队列与交换机的所有路由键绑定，而非单个路由键
         */
        void queueUnBind(const std::string &ename, const std::string &qname)
        {
            std::string rid = UUIDHelper::uuid();
            queueUnBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }

        /**
         * @brief 发布消息：向指定交换机发送消息，由交换机路由到匹配的队列
         * 消息发布流程：
         * 1. 构造basicPublishRequest，包含消息体、交换机名称、消息属性
         * 2. 发送消息后等待服务器确认（确保消息已被服务器接收）
         * @param ename 交换机名称（为空时使用默认交换机）
         * @param bp 消息属性（可为nullptr，包含消息ID、投递模式、路由键等）
         * @param body 消息体（原始字符串，通常为业务序列化后的数据）
         * 投递模式说明：
         * - 1：非持久化（服务器重启后消息丢失）
         * - 2：持久化（服务器会将消息写入磁盘，重启后可恢复）
         */
        void basicPublish(
            const std::string &ename,
            const BasicProperties *bp,
            const std::string &body)
        {
            std::string rid = UUIDHelper::uuid();
            basicPublishRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_body(body); // 设置消息体
            req.set_exchange_name(ename);
            // 复制消息属性（若存在）
            if (bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());                       // 消息唯一ID
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode()); // 投递模式
                req.mutable_properties()->set_routing_key(bp->routing_key());     // 用于路由的键
            }
            _codec->send(_conn, req);
            waitResponse(rid); // 等待服务器确认消息已接收
            return;
        }

        /**
         * @brief 确认消息已处理：通知服务器消息已被消费，可从队列中删除
         * 消息确认机制：
         * - 当auto_ack为false时，消费者必须显式调用basicAck确认消息处理完成
         * - 服务器收到确认后，才会从队列中删除该消息（避免消息丢失）
         * @param msgid 要确认的消息ID（从消费到的消息属性中获取）
         * 异常处理：
         * - 若当前信道没有消费者（_consumer为空），打印日志并返回
         */
        void basicAck(const std::string &msgid)
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("消息确认时，找不到消费者信息！"); // 使用自定义日志宏输出调试信息
                return;
            }
            std::string rid = UUIDHelper::uuid();
            basicAckRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname); // 关联消费者订阅的队列
            req.set_message_id(msgid);            // 要确认的消息ID
            _codec->send(_conn, req);
            waitResponse(rid); // 等待服务器确认已处理
            return;
        }

        /**
         * @brief 取消消费者订阅：终止当前信道的消息消费，释放相关资源
         * 流程：
         * 1. 若没有消费者，则直接返回
         * 2. 向服务器发送取消请求，包含队列名称和消费者标签
         * 3. 等待服务器响应后，重置消费者指针（_consumer.reset()）
         * 注意：析构函数会调用此方法，确保资源正确释放
         */
        void basicCancel()
        {
            if (_consumer.get() == nullptr)
            {
                return;
            }
            std::string rid = UUIDHelper::uuid();
            basicCancelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname); // 订阅的队列名称
            req.set_consumer_tag(_consumer->tag); // 消费者唯一标签
            _codec->send(_conn, req);
            waitResponse(rid); // 等待服务器确认取消
            _consumer.reset(); // 释放消费者对象
            return;
        }

        /**
         * @brief 订阅队列消息：创建消费者，接收服务器推送的消息并通过回调处理
         * 消费模式：
         * - 采用推模式（push）：服务器有消息时主动推送给消费者
         * - 每个信道同一时间只能订阅一个队列（_consumer为单例）
         * @param consumer_tag 消费者标签（用于标识消费者，可自定义或由服务器生成）
         * @param queue_name 要订阅的队列名称
         * @param auto_ack 是否自动确认（true：服务器推送后自动确认；false：需手动调用basicAck）
         * @param cb 消息处理回调函数（当收到消息时被调用）
         * @return 订阅成功返回true，否则返回false
         * 限制：
         * - 若当前信道已存在消费者，打印日志并返回false
         */
        bool basicConsume(
            const std::string &consumer_tag,
            const std::string &queue_name,
            bool auto_ack,
            const ConsumerCallback &cb)
        {
            // 检查是否已存在消费者（一个信道只能有一个消费者）
            if (_consumer.get() != nullptr)
            {
                DLOG("当前信道已订阅其他队列消息！");
                return false;
            }
            std::string rid = UUIDHelper::uuid();
            basicConsumeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(queue_name);
            req.set_consumer_tag(consumer_tag);
            req.set_auto_ack(auto_ack);
            _codec->send(_conn, req);

            // 等待服务器响应，确认订阅是否成功​
            basicCommonResponsePtr resp = waitResponse(rid);
            if (resp->ok() == false)
            {
                DLOG("添加订阅失败！");
                return false;
            }
            // 创建消费者对象，保存订阅信息和回调函数​
            _consumer = std::make_shared<Consumer>(consumer_tag, queue_name, auto_ack, cb);
            return true;
        }

    public:
        // 连接收到基础响应后，向hash_map中添加响应
        void putBasicResponse(const basicCommonResponsePtr &resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert(std::make_pair(resp->rid(), resp));
            _cv.notify_all();
        }

        // 连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进行消息处理
        void consume(const basicConsumeResponsePtr &resp)
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("消息处理时，未找到订阅者信息！");
                return;
            }
            if (_consumer->tag != resp->consumer_tag())
            {
                DLOG("收到的推送消息中的消费者标识，与当前信道消费者标识不一致！");
                return;
            }
            _consumer->callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
        }

    private:
        /**
         * @brief 等待指定请求ID对应的服务器响应
         * @param rid 待等待的请求ID（与发送请求时的rid对应）
         * @return 服务器返回的对应响应消息智能指针
         */
        basicCommonResponsePtr waitResponse(const std::string &rid)
        {
            // 构造unique_lock并加锁，自动管理锁的生命周期（离开作用域自动解锁）
            std::unique_lock<std::mutex> lock(_mutex);
            // 等待条件变量，直到lambda表达式返回true（即找到对应响应）
            _cv.wait(lock, [&rid, this]()
                     { return _basic_resp.find(rid) != _basic_resp.end(); });
            // 从响应映射表中获取对应响应
            basicCommonResponsePtr basic_resp = _basic_resp[rid];
            // 移除已处理的响应，防止重复获取
            _basic_resp.erase(rid);
            return basic_resp;
        }

    private:
        std::string _cid;                   // 信道唯一标识符（UUID），用于在TCP连接中区分不同信道
        muduo::net::TcpConnectionPtr _conn; // 关联的TCP连接指针，所有消息通过该连接发送
        ProtobufCodecPtr _codec;            // Protobuf编解码器指针，负责消息的序列化与反序列化
        Consumer::ptr _consumer;            // 消费者对象指针，仅在调用basicConsume后非空，存储订阅相关信息
        std::mutex _mutex;                  // 互斥锁，保护所有共享资源的并发访问（如_basic_resp、_consumer等）
        std::condition_variable _cv;        // 条件变量，配合_mutex使用，实现waitResponse的阻塞等待机制
        // 存储请求ID到响应消息的映射表，用于暂存服务器返回的响应，等待waitResponse处理
        std::unordered_map<std::string, basicCommonResponsePtr> _basic_resp;
    };

    /**
     * @brief 信道管理器类，负责对所有信道进行统一管理（创建、查找、删除）
     * 设计目的：在多信道场景下，提供线程安全的信道管理接口，通过信道ID快速定位对应的Channel对象
     * 核心特性：线程安全，所有方法均通过互斥锁保护共享资源_channels
     */
    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>; // 定义ChannelManager的智能指针类型，方便生命周期管理

        /**
         * @brief 默认构造函数，初始化管理器内部状态
         */
        ChannelManager() {}

        /**
         * @brief 创建新的信道对象，并将其加入管理列表
         * 实现流程：
         * 1. 加锁确保线程安全
         * 2. 构造新的Channel对象（传入TCP连接和编解码器）
         * 3. 将新信道以<cid, 信道指针>的形式存入_channels映射表
         * 4. 返回新创建的信道智能指针
         * @param conn 已建立的TCP连接指针
         * @param codec Protobuf编解码器指针
         * @return 新创建的信道智能指针
         */
        Channel::ptr create(const muduo::net::TcpConnectionPtr &conn,
                            const ProtobufCodecPtr &codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);                 // 加锁，防止并发创建信道时的竞态条件
            auto channel = std::make_shared<Channel>(conn, codec);     // 构造新信道对象
            _channels.insert(std::make_pair(channel->cid(), channel)); // 将信道加入管理表
            return channel;
        }

        /**
         * @brief 从管理列表中移除指定ID的信道
         * 注意事项：
         * 1. 仅移除_channels中的映射关系，信道对象的销毁由智能指针的引用计数决定
         * 2. 通常在信道关闭或TCP连接断开时调用，确保管理列表与实际信道状态一致
         * @param cid 要移除的信道ID
         */
        void remove(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex); // 加锁，确保线程安全
            _channels.erase(cid);                      // 从映射表中删除指定信道
        }

        /**
         * @brief 根据信道ID查找对应的信道对象
         * 应用场景：当服务器消息到达时，通过消息中的cid查找对应的Channel进行处理
         * @param cid 要查找的信道ID
         * @return 找到则返回对应信道的智能指针，否则返回空指针
         */
        Channel::ptr get(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex); // 加锁，确保查找过程的线程安全
            auto it = _channels.find(cid);             // 在映射表中查找指定cid
            if (it == _channels.end())
            {
                return Channel::ptr(); // 未找到，返回空指针
            }
            return it->second; // 找到，返回信道智能指针
        }

    private:
        std::mutex _mutex; // 互斥锁，保护_channels的并发访问
        // 存储信道ID到信道对象的映射表，支持O(1)时间复杂度的查找、插入和删除操作
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}
#endif